package com.sun.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * 单个消费者消费单个Topic
 */
@Slf4j
@Component
public class KafkaConsumerListener {

    @KafkaListener(topics = { "simple1" })
    public void listenRecord(ConsumerRecord<?, ?> record) {
        log.info("record = {}", record);
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());

        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            log.info("topic = {}, partition = {}, message = {}", record.topic(), record.partition(), message);
        }
    }

    /**
     * 必须要修改以下配置才能生效
     * spring.kafka.consumer.enable-auto-commit=false
     * spring.kafka.listener.ack-mode=manual
     */
    @KafkaListener(topics = { "simple2" })
    public void listenRecordWhitAck(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            log.info("record = {}", record);
            if (kafkaMessage.isPresent()) {
                Object message = kafkaMessage.get();
                log.info("topic = {}, partition = {}, message = {}", record.topic(), record.partition(), message);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            // 手动提交 offset
            ack.acknowledge();
        }
    }

}
